package org.jpush.core;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;

import org.jpush.core.monitor.MonitorAcceptor;
import org.jpush.core.monitor.MonitorsChannelManager;
import org.jpush.packages.OutgoingPacket;
import org.jpush.utils.Utilities;
import org.jpush.utils.Version;

public class Server
{
	private String serverInfos;

    // User defined
	private Protocol protocol;
	private ConnectionEventListener connectionEventListener;
	private short listeningPort;
	private int workerCount;

    // Remote monitoring
	private boolean isMonitorEnabled;
	private boolean isProfilingEnabled;
	private int monitorPort;
	private int samplingRate; // how many seconds
	private String password;

	private int uGCPeriod;
	private int nSecsGCCount;
	private int nSecsPerfObservCount;

	// Used by monitor(see DoPeriodicJobs()). cancel it when stop()
	private Timer timer;

	//Listeners with their options :
	private Map<Integer, Acceptor> listenersMap;

	// Components of this server
	private ChannelManager broadcastManager;
	private ChannelStreamerManager broadcastStreamerManager;
	private ConnectionFactory clientFactory;
	private PhysicalConnectionManager channelFactory;
	private Dispatcher dispatcher;
	private Demultiplexor demultiplexor;
	private PhysicalConnectionEventQueue connectionEventQueue;
	private ServerStats serverStats;
	
	// Components of this server monitor
	private MonitorAcceptor monitorAcceptor;
	private MonitorsChannelManager monitorsBroadcastManager;
	
	private int serverSocketOption;
	private int connectionEventQueueOption;
	
	public Server()
	{
	    protocol = null;
	    connectionEventListener = null;
	    listeningPort = 2010;

	    workerCount = Utilities.getProcessorsCount() * 2;

	    // Monitor setting
	    isMonitorEnabled = false;
	    isProfilingEnabled = false;
	    monitorPort = 2011;
	    samplingRate = 10;
	    uGCPeriod = 10;
	    password = "";
	    
	    // Components' initiation
	    serverStats = new ServerStats(this);
	    broadcastManager = new ChannelManager(this);
	    broadcastStreamerManager = new ChannelStreamerManager(broadcastManager);
	    channelFactory = new PhysicalConnectionManager(this);
	    clientFactory = new ConnectionFactory(this);
	    dispatcher = new Dispatcher(this);
	    demultiplexor = new Demultiplexor(this);

	    monitorAcceptor = new MonitorAcceptor(this);
	    monitorsBroadcastManager = new MonitorsChannelManager();
	    
	    serverSocketOption = JPushComponentFactory.DEFAULT_SERVER_SOCKET;
	    connectionEventQueueOption = JPushComponentFactory.DEFAULT_PHYSICAL_CONNECTION_EVENT_QUEUE;
	}

	public void setServeInfos(String serverInfos)
	{
	    this.serverInfos = serverInfos;
	}

	public String getServerInfos()
	{
	    return serverInfos;
	}

	public void setProtocol(Protocol protocol)
	{
	    this.protocol = protocol;
	}

	public Protocol getProtocol()
	{
	    return this.protocol;
	}

	public void setConnectionEventListener(ConnectionEventListener clientEventListener)
	{
	    this.connectionEventListener = clientEventListener;
	}

	public boolean createListener(int port)
	{
	    return createListener(port, null);
	}

	public boolean createListener(int port, ListenerOptions options)
	{
		Integer key = new Integer(port);
		if (listenersMap.containsKey(key))
			return false;

	    Acceptor acceptor = new Acceptor(this);
	    acceptor.setListeningPort(port);

	    if (options != null)
	    {
	    	acceptor.setOptions(options);
	    }

	    listenersMap.put(key, acceptor);
	    return true;
	}

	public void enableRemoteMonitor(int port, String password)
	{
	    this.monitorPort = port;
	    this.password = password;
	    isMonitorEnabled = true;
	    monitorAcceptor.setListeningPort(monitorPort);
	}

	public void enableProfiling(int samplingRate)
	{
	    if (samplingRate >= 0)
	    {
	        this.samplingRate = samplingRate;
		    isProfilingEnabled = true;
	    }
	}

	public void disableProfiling()
	{
	    isProfilingEnabled = false;
	}

	public boolean getProfilingStatus()
	{
	    return isProfilingEnabled;
	}


	public void setMaxConnections(int nMaxConnections)
	{
	    channelFactory.setMaxAllowedConnections(nMaxConnections);
	}


	public void setLoginExpiryDuration(int uLoginExpireDuration)
	{
	    channelFactory.setLoginExpiryDuration(uLoginExpireDuration);
	}

	public void setWorkerCount(int workerCount)
	{
	    this.workerCount = workerCount;
	}

	public void registerService(int serviceId, Service pService, String serviceName)
	{
	    dispatcher.registerService(serviceId, pService, serviceName);
	}

	public void start()
	{
	    if(connectionEventListener == null)
	    {
	        throw new RuntimeException("Fatal : no ClientEventListener instance was set");
	    }

	    assignDefaultProtocol();
	    
	    // Create connection event queue
	    connectionEventQueue = JPushComponentFactory.getPhysicalConnectionEventQueue(
	    		connectionEventQueueOption);
	    
	    if (!connectionEventQueue.create())
	    {
	        throw new RuntimeException("Fatal : failed to create Event Queue");
	    }
	    
	    if (!demultiplexor.start())
	    {
	    	connectionEventQueue.destroy();
	        throw new RuntimeException("Fatal : failed to start Demultiplexor");
	    }

	    if (!startListeners())
	    {
	        stopListeners();
	        demultiplexor.stop();
	        connectionEventQueue.destroy();
	    }

	    if (isMonitorEnabled && !monitorAcceptor.startListening())
	    {
	        stopListeners();
	        throw new RuntimeException("Internal problem : unable to start monitoring acceptor");
	    }

	    broadcastStreamerManager.start();

	    startMonitorService();

	}
	public void stop()
	{		
		if (timer != null)
		{
			timer.cancel();
			timer = null;
		}

	    if (isMonitorEnabled)
	    {
	        monitorAcceptor.stopListening();
	    }

	    stopListeners();

        demultiplexor.stop();

	    broadcastManager.disposeAllPackets();

        connectionEventQueue.destroy();
	}

	public void pause()
	{
	    stopListeners();
	}

	public void resume()
	{
	    startListeners();
	}

	public ConnectionEventListener getConnectionEventListener()
	{
	    return connectionEventListener;
	}

	public int getSamplingRate()
	{
	    return samplingRate;
	}

	public String getMonitorPassword()
	{
	    return password;
	}


	public int getWorkerCount()
	{
	    return workerCount;
	}

	public void getVersion(Version ver)
	{
	    ver.major = 1;
	    ver.minor = 0;
	    ver.monitorProtocolVer = 1;
	}

    public void disposeOutgoingPacket(OutgoingPacket pPacket)
    {
    	// Nothing will be done
    }

	private void startMonitorService()
	{	
		// 
		timer = new Timer();
		timer.schedule(new TimerTask()
		{
			public void run()
			{
				DoPeriodicJobs();
			}
			
		}, 1000L, 1000L);
	}

	private void DoPeriodicJobs()
	{
	    nSecsGCCount++;
	    nSecsPerfObservCount++;

	    if(nSecsGCCount == uGCPeriod)
	    {
	        nSecsGCCount=0;
	        ConnectionGC.getGC().activate();
	        channelFactory.closeNonLogged();
	    }

	    if(nSecsPerfObservCount == samplingRate && isProfilingEnabled)
	    {
	        nSecsPerfObservCount=0;
	        monitorsBroadcastManager.pushPacket(serverStats.getPerformancePacket(), "stats");
	    }
	}
	//
	private boolean startListeners()
	{		
		for (Acceptor acceptor : listenersMap.values())
		{
			if (!acceptor.startListening())
				return false;
		}
	    return true;
	}


	private void stopListeners()
	{		
		for (Acceptor acceptor : listenersMap.values())
		{
			acceptor.stopListening();
		}
	}

	/**
	 * Assign the protocol of this server to each <code>Acceptor</code>
	 * as its default protocol
	 */
	private void assignDefaultProtocol()
	{		
		Acceptor acceptor;
		for (Entry<Integer, Acceptor> entry : listenersMap.entrySet())
		{
			acceptor = entry.getValue();
			if (acceptor.getOptions() != null && acceptor.getOptions().getProtocol() == null)
			{
				if (protocol == null)
					throw new RuntimeException("FATAL : no protocol instance for acceptor on port : "+ entry.getKey());
				
				// Set protocol
				acceptor.getOptions().setProtocol(protocol);
			}
		}
	}

	/**
	 * Encode the <code>OutgoingPacket</code> with the protocol of each
	 * <code>Acceptor</code> one by one
	 * @param outgoingPacket
	 */
	public void encodeOnAllProtocols(OutgoingPacket outgoingPacket)
	{		
		for (Acceptor acceptor : listenersMap.values())
		{
			if (acceptor.getOptions() != null &&
					acceptor.getOptions().getProtocol() != null)
			{
				acceptor.getOptions().getProtocol().encodeOutgoingPacket(outgoingPacket);
			}
		}
	}
	
	public short getListeningPort()
	{
		return this.listeningPort;
	}
	
	public ServerStats getServerStats()
	{
		return serverStats;
	}
	
	public ChannelManager getBroadcastManager()
	{
		return broadcastManager;
	}
	
	public ChannelStreamerManager getBroadcastStreamerManager()
	{
		return broadcastStreamerManager;
	}
	
	public MonitorAcceptor getMonitorAcceptor()
	{
		return monitorAcceptor;
	}
	
	public PhysicalConnectionManager getChannelFactory()
	{
		return channelFactory;
	}
	
	
	public Dispatcher getDispatcher()
	{
		return this.dispatcher;
	}
	
	public MonitorsChannelManager getMonitorsBroadcastManager()
	{
		return this.monitorsBroadcastManager;
	}
	
	public ConnectionFactory getClientFactory()
	{
		return this.clientFactory;
	}

	/**
	 * @param command A null terminating string containing the received user command.
	 * @param sbOut   An 256-length buffer. Write a null terminating string to this buffer 
	 *                and return the value of true so the framework sends it as a reply.
	 */
	public boolean handleMonitorRequest(String command, StringBuffer sbOut)
	{
		return false;
	}
	
	/**
	 * The option for creating JPushServerSocket object.
	 * see <code>JPushServerSocketFactory</code>
	 * @return
	 */
	public int getServerSocketOption()
	{
		return this.serverSocketOption;
	}
	
	public void setServerSocketOption(int serverSocketOption)
	{
		this.serverSocketOption = serverSocketOption;
	}

	public int getConnectionEventQueueOption()
	{
		return this.connectionEventQueueOption;
	}
	
	public void setConnectionEventQueueOption(int connectionEventQueueOption)
	{
		this.connectionEventQueueOption = connectionEventQueueOption;
	}

	public PhysicalConnectionEventQueue getConnectionEventQueue()
	{
		return this.connectionEventQueue;
	}
}
